package shoppingAnalysis;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import java.io.IOException;
import java.net.URI;

//用Mapreduce将列与列之间的分隔改为逗号，将结果写回hdfs
public class Task1 {

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);// 设置环境参数
		job.setJarByClass(Task1.class);// 设置整个程序的类名

		job.setMapperClass(ShopMapper.class);
		job.setReducerClass(ShopReducer.class);

		// reduce 输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		// 设置自定义的格式化输出
		job.setOutputFormatClass(ShopFileOutputFormat.class);
		job.setNumReduceTasks(1);

		FileInputFormat.setInputPaths(job,
				new Path("hdfs://192.168.3.101:9000/ShoppingAnalysis/input/raw_user_table.txt"));
		// 如果文件系统已存在输出文件夹则删除
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.3.101:9000"), conf, "roger");
		if (fs.exists(new Path("/ShoppingAnalysis/output"))) {
			fs.delete(new Path("/ShoppingAnalysis/output"), true);
		}

		// 注意这里是WCFileOutPutFormat，不是FileOutPutFormat，下面同样注意
		ShopFileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.3.101:9000/ShoppingAnalysis/output"));

		/* 消除系统的默认输出文件，不过因为文件的输出格式变了，不再是FileOutPutFormat，所以这里也要改成相对应的文件输出格式 */
		LazyOutputFormat.setOutputFormatClass(job, ShopFileOutputFormat.class);

		job.getConfiguration().setStrings("mapreduce.reduce.shuffle.memory.limit.percent", "0.1");

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}

class ShopMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		context.write(new Text(key.toString()), value);

	}
}

class ShopReducer extends Reducer<Text, Text, Text, Text> {

	private MultipleOutputs<Text, Text> multipleOutputs;

	@Override
	protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		multipleOutputs = new MultipleOutputs<Text, Text>(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		String line = "";
		for (Text text : values) {
			line += text.toString();
		}
		String[] words = line.toString().split("\\s+");
		String tmp = words[0];
		for (int i = 1; i < words.length; i++) {
			tmp += "," + words[i];
		}
		// multipleOutputs.write(Key输出，value输出，输出的文件名)
		multipleOutputs.write(new Text(tmp), new Text(""), "result1");

	}

	@Override
	protected void cleanup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		multipleOutputs.close();
	}
}
